package xin.marcher.wind.migrate.component.canal;

import cn.hutool.core.collection.CollUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import xin.marcher.framework.common.util.spring.SpringContextUtil;
import xin.marcher.wind.migrate.constants.MigrateConstant;
import xin.marcher.wind.migrate.domain.BinlogData;
import xin.marcher.wind.migrate.domain.entity.EtlBinlogConsumeRecordDO;
import xin.marcher.wind.migrate.enums.BinlogType;
import xin.marcher.wind.migrate.enums.ConsumerStatus;
import xin.marcher.wind.migrate.mapper.EtlBinlogConsumeRecordMapper;
import xin.marcher.wind.migrate.migrate.LocalQueue;
import xin.marcher.wind.migrate.util.BinlogUtils;

import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * binlog 消息拉取任务（只拉不提交）
 */
@Slf4j
public class CanalPullRunner implements Runnable {

    /**
     * 消息主题
     */
    private final String topic;

    /**
     * rocketmq 的 nameServer 地址
     */
    private final String mqNameServer;

    /**
     * binlog 消息同步消费记录表 Mapper
     */
    private final EtlBinlogConsumeRecordMapper consumeRecordMapper;

    /**
     * 消息拉取任务构造方法
     *
     * @param topic         消息主题
     * @param mqNameServer rocketmq 的 nameServer 地址
     */
    public CanalPullRunner(String topic, String mqNameServer) {
        this.topic = topic;
        this.mqNameServer = mqNameServer;
        this.consumeRecordMapper = SpringContextUtil.getBean(EtlBinlogConsumeRecordMapper.class);
    }

    @Override
    public void run() {
        pullRun();
    }

    /**
     * 执行消息拉取
     */
    private void pullRun() {
        try {
            DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(MigrateConstant.MQ_BINLOG_PULL_GROUP);
            litePullConsumer.setAutoCommit(false);
            litePullConsumer.setNamesrvAddr(mqNameServer);
            litePullConsumer.subscribe(topic, "*");
            litePullConsumer.start();

            try {
                while (true) {
                    // 拉取未消费消息, poll 默认 32 条
                    List<MessageExt> messageExts = litePullConsumer.poll();
                    if (CollUtil.isNotEmpty(messageExts)) {
                        for (MessageExt messageExt : messageExts) {
                            byte[] body = messageExt.getBody();
                            String msg = new String(body);
                            // 记录 queueId 和 offset
                            int queueId = messageExt.getQueueId();
                            long offset = messageExt.getQueueOffset();
                            String topic = messageExt.getTopic();

                            // 判断该消息是否已经存在消费记录，如果存在则跳过执行
                            EtlBinlogConsumeRecordDO existsRecord = consumeRecordMapper.getExistsRecord(queueId, offset, topic);
                            if (null == existsRecord) {
                                // 新增消费记录
                                processNewMsg(messageExt, msg);
                            } else {
                                // 处理已经存在的消费记录
                                processExistsRecord(litePullConsumer, msg, existsRecord);

                            }
                        }
                    } else {
                        Thread.sleep(5000);
                    }
                }
            } finally {
                litePullConsumer.shutdown();
            }
        } catch (InterruptedException | MQClientException e) {
            try {
                // 假设要拉取消息的主题还不存在，则会抛出异常，这种情况下休眠五秒再重试
                Thread.sleep(5000);
                pullRun();
            } catch (InterruptedException ignored) {

            }
        }
    }

    /**
     * 处理新的消息
     *
     * @param messageExt mq 消息对象
     * @param msg        消息内容
     */
    private void processNewMsg(MessageExt messageExt, String msg) {
        try {
            BinlogData binlogData = BinlogUtils.getBinlogDataMap(msg);
            if (Objects.isNull(binlogData)) {
                return;
            }
            Boolean targetOperateType = BinlogType.INSERT.getValue().equals(binlogData.getOperateType())
                    || BinlogType.DELETE.getValue().equals(binlogData.getOperateType())
                    || BinlogType.UPDATE.getValue().equals(binlogData.getOperateType());
            if (!targetOperateType || null == binlogData.getDataMap()) {
                return;
            }

            // 每一条 binlog，都是提交到本地队列里去，依托队列，实现异步化的处理
            EtlBinlogConsumeRecordDO consumeRecord = buildEtlBinlogConsumeRecordDO(messageExt);
            consumeRecordMapper.insert(consumeRecord);

            /*
            * 本地队列
            * binlog 数据会基于本地队列去提交，触发异步化的处理
            * 本条消息 offset 是不会提交给 broker 的，必须等到异步化处理玩这个 binlog 之后才能去提交这条消息到 broker 里去
            */
            LocalQueue.getInstance().submit(binlogData, consumeRecord);
        } catch (Exception e) {
            log.error("新增消费记录失败", e);
        }
    }

    private static EtlBinlogConsumeRecordDO buildEtlBinlogConsumeRecordDO(MessageExt messageExt) {
        EtlBinlogConsumeRecordDO consumeRecord = new EtlBinlogConsumeRecordDO();
        consumeRecord.setQueueId(messageExt.getQueueId());
        consumeRecord.setOffset(messageExt.getQueueOffset());
        consumeRecord.setTopic(messageExt.getTopic());
        consumeRecord.setBrokerName(messageExt.getBrokerName());
        consumeRecord.setConsumeStatus(ConsumerStatus.NOT_CONSUME.getValue());
        consumeRecord.setCreateTime(new Date());
        return consumeRecord;
    }

    /**
     * 处理已经存在的消费记录
     *
     * @param litePullConsumer mq 消费者
     * @param msg              消息内容
     * @param existsRecord     已经存在的消费记录
     */
    private void processExistsRecord(DefaultLitePullConsumer litePullConsumer, String msg, EtlBinlogConsumeRecordDO existsRecord) {
        // 已经存在的消费记录状态为已提交，说明 mq 里的对应消息修改提交状态失败了
        // （rocketmq源码里手动提交消息时，如果失败了只会记录日志不会抛出异常，因此这里必须再次尝试提交消息防止 mq 中未处理的消息和实际情况不符）
        try {
            if (ConsumerStatus.COMMITTED.getValue().equals(existsRecord.getConsumeStatus())) {
                litePullConsumer.seek(new MessageQueue(existsRecord.getTopic(), existsRecord.getBrokerName(), existsRecord.getQueueId()), existsRecord.getOffset());
                // 这一步必须
                List<MessageExt> messageExts = litePullConsumer.poll();
                // 再次提交已消费的消息
                litePullConsumer.commitSync();
            } else {
                BinlogData binlogData = BinlogUtils.getBinlogDataMap(msg);
                if (null == binlogData) {
                    return;
                }
                LocalQueue.getInstance().submit(binlogData, existsRecord);
            }
        } catch (Exception e) {
            log.error("消息重新消费失败", e);
        }
    }
}
